package jrain.fw.mq.ons;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.OrderProducer;

import jrain.fw.core.cfg.CfgService;
import jrain.fw.core.mq.MqProducerService;
import jrain.fw.core.service.DestroyService;
import jrain.fw.core.service.InitService;
import jrain.fw.core.service.Service;
import jrain.fw.core.service.ServiceConst;
import jrain.fw.core.utils.FwUtils;
import jrain.utils.collection.RMap;
import jrain.utils.consts.BaseConst;
import jrain.utils.lang.LangUtils;
import jrain.utils.lang.StringUtils;

@Service(name = "ons", group = ServiceConst.S_MQ_PRODUCER_SERVICE, module = ServiceConst.M_MQ)
public class MqOnsProducerService implements MqProducerService, InitService, DestroyService {

	private static final AtomicBoolean init = new AtomicBoolean(false);
	private Producer producer;
	private OrderProducer orderProducer;

	public String send(String topic, String tag, String data, String key) {
		Message message = new Message(topic, tag, data.getBytes(BaseConst.UTF8));
		if (key != null && key.length() != 0) {
			message.setKey(key);
		}
		return producer.send(message).getMessageId();
	}

	public String sendOrder(String topic, String tag, String data, String key) {
		return sendOrder(topic, tag, data, key, "global");
	}

	public String sendOrder(String topic, String tag, String data, String key, String shardingKey) {
		Message message = new Message(topic, tag, data.getBytes(BaseConst.UTF8));
		if (key != null && key.length() != 0) {
			message.setKey(key);
		}
		return orderProducer.send(message, shardingKey).getMessageId();
	}

	public void init() {
		if (!init.getAndSet(true)) {
			CfgService cfgService = FwUtils.getCfgService();
			RMap<String, Object> cfgMap = cfgService.getCfgMap(ServiceConst.M_MQ, null);
			if (cfgMap != null) {
				boolean enable = LangUtils.parseBoolean(cfgMap.get("mq.enable"), false);
				if (enable) {
					Properties properties = new Properties();
					String producerId = StringUtils.trimNull(cfgMap.get("mq.producer.id"));
					String accessKey = StringUtils.trimNull(cfgMap.get("mq.access.key"));
					String secretKey = StringUtils.trimNull(cfgMap.get("mq.secret.key"));
					String onsAddr = StringUtils.trimNull(cfgMap.get("mq.ons.addr"));
					properties.put(PropertyKeyConst.ProducerId, producerId);
					// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
					properties.put(PropertyKeyConst.AccessKey, accessKey);
					// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
					properties.put(PropertyKeyConst.SecretKey, secretKey);
					// 设置 TCP 接入域名（此处以公共云生产环境为例）
					properties.put(PropertyKeyConst.ONSAddr, onsAddr);
					orderProducer = ONSFactory.createOrderProducer(properties);
					producer = ONSFactory.createProducer(properties);
					// 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可。
					orderProducer.start();
					producer.start();
				}
			}
		}
	}

	@Override
	public void destroy() {
		if (init.get()) {
			if (producer != null) {
				try {
					producer.shutdown();
				} catch (Throwable t) {
				}
			}
			if (orderProducer != null) {
				try {
					orderProducer.shutdown();
				} catch (Throwable t) {
				}
			}
			init.set(false);
		}
	}

}
